package com.dtp.adapter.rocketmq;

import com.dtp.adapter.common.AbstractDtpAdapter;
import com.dtp.common.ApplicationContextHolder;
import com.dtp.common.properties.DtpProperties;
import com.dtp.common.util.ReflectionUtil;
import com.dtp.core.support.ExecutorWrapper;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.commons.collections.MapUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService;
import org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService;
import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;

import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * RocketMQ动态线程池适配器类
 *
 * @author yanhom
 * @since 1.0.6
 *
 * @remark muse
 */
@SuppressWarnings("all")
@Slf4j
public class RocketMqDtpAdapter extends AbstractDtpAdapter {
    private static final String NAME = "rocketMqTp";
    private static final String CONSUME_EXECUTOR_FIELD_NAME = "consumeExecutor";

    @Override
    public void refresh(DtpProperties dtpProperties) {
        refresh(NAME, dtpProperties.getRocketMqTp(), dtpProperties.getPlatforms());
    }

    @Override
    protected void initialize() {
        super.initialize();
        adaptConsumerExecutors(); // 获得消费者线程池，并缓存到executors中
        adaptProducerExecutors(); // 获得生产者线程池，并缓存到executors中
        log.info("DynamicTp adapter, rocketMq consumer and producer executors init end, executors: {}", executors);
    }

    /** 获得消费者线程池，并缓存到executors中 */
    public void adaptConsumerExecutors() {
        // 如果不存在这个类型的bean，则直接返回
        val beans = ApplicationContextHolder.getBeansOfType(DefaultRocketMQListenerContainer.class);
        if (MapUtils.isEmpty(beans)) return;

        beans.forEach((k, v) -> {
            // 如果不存在pushConsumer，则直接返回
            DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) v;
            DefaultMQPushConsumer consumer = container.getConsumer();
            val pushConsumer = (DefaultMQPushConsumerImpl) ReflectionUtil.getFieldValue(DefaultMQPushConsumer.class, "defaultMQPushConsumerImpl", consumer);
            if (Objects.isNull(pushConsumer)) return;

            // 获得消费者线程池executor
            ThreadPoolExecutor executor = null;
            val consumeMessageService = pushConsumer.getConsumeMessageService();
            if (consumeMessageService instanceof ConsumeMessageConcurrentlyService)
                executor = (ThreadPoolExecutor) ReflectionUtil.getFieldValue(ConsumeMessageConcurrentlyService.class, CONSUME_EXECUTOR_FIELD_NAME, consumeMessageService);
            else if (consumeMessageService instanceof ConsumeMessageOrderlyService)
                executor = (ThreadPoolExecutor) ReflectionUtil.getFieldValue(ConsumeMessageOrderlyService.class, CONSUME_EXECUTOR_FIELD_NAME, consumeMessageService);

            String cusKey = container.getConsumerGroup() + "#" + container.getTopic(); // 拼装线程池名称：threadPoolName
            if (Objects.nonNull(executor)) {
                val executorWrapper = new ExecutorWrapper(cusKey, executor);
                initNotifyItems(cusKey, executorWrapper); // 初始化报警通知
                executors.put(cusKey, executorWrapper); // 将线程池缓存起来
            }
        });
    }

    /** 获得生产者线程池，并缓存到executors中 */
    public void adaptProducerExecutors() {
        // 如果不存在这个类型的bean，则直接返回
        val beans = ApplicationContextHolder.getBeansOfType(DefaultMQProducer.class);
        if (MapUtils.isEmpty(beans)) return;

        beans.forEach((k, v) -> {
            // 如果不存在producer，则直接返回
            DefaultMQProducer defaultMQProducer = (DefaultMQProducer) v;
            val producer = (DefaultMQProducerImpl) ReflectionUtil.getFieldValue(DefaultMQProducer.class, "defaultMQProducerImpl", defaultMQProducer);
            if (Objects.isNull(producer)) return;

            // 获得生产者线程池executor
            ThreadPoolExecutor executor = (ThreadPoolExecutor) producer.getAsyncSenderExecutor();

            String proKey = defaultMQProducer.getProducerGroup() + "#" + defaultMQProducer.getCreateTopicKey(); // 拼装线程池名称：threadPoolName
            if (Objects.nonNull(executor)) {
                val executorWrapper = new ExecutorWrapper(proKey, executor);
                initNotifyItems(proKey, executorWrapper); // 初始化报警通知
                executors.put(proKey, executorWrapper); // 将线程池缓存起来
            }
        });
    }
}
